package com.bigdata.core.example

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object PvAndUv {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("pvuv")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val lines = sc.textFile("data/pvuvdata")
    //pv, 统计每个url出现的次数
    val pvResult: RDD[(String, Int)] = lines.map(line => {
      (line.split("\t")(5), 1)
    }).reduceByKey(_ + _)

    pvResult.sortBy(kv => {
      kv._2
    }, false).foreach(println)

    println("pv count end....")

    //uv 对ip地址去重 并统计每个url出现的次数
    // ip和 Url 组合去重
    val uvRes1: RDD[(String, Int)] = lines.map(line => {
      val arr: Array[String] = line.split("\t")
      (arr(0) + "-" + arr(5), 1)
    }).distinct()


    val uvRes2: RDD[(String, Int)] = uvRes1.map(kv => {
      (kv._1.split("-")(1), 1)
    }).reduceByKey(_ + _).sortBy(_._2, false)
    uvRes2.foreach(println)

    println("uv count end....")

    // 计算每个网址 最活跃的地区和地区的活跃人数
    //    val groupTrans: RDD[(String, Iterable[String])] = lines.map(data => {
    //      (data.split("\t")(5), data.split("\t")(1))
    //    }).groupByKey()
    //
    //    groupTrans.map(data => {
    //      val site: String = data._1
    //      val localIter: Iterator[String] = data._2.iterator
    //      val localMap: mutable.Map[String, Int] = mutable.Map[String, Int]()
    //
    //      // 遍历地区
    //      localIter.foreach((elem: String) => {
    //        // 统计地区出现的次数
    //        if (localMap.contains(elem)) {
    //          localMap.put(elem, localMap.get(elem).get + 1)
    //        } else {
    //          localMap.put(elem, 1)
    //        }
    //      })
    //      // 默认是升序, 取反是倒序
    //      val newList: List[(String, Int)] = localMap.toList.sortBy(-_._2)
    //      (site, newList)
    //
    //    }).foreach(println)

  }
}
